package com.fengye.redisson.demo.message;

import com.fengye.redisson.demo.dto.MessageDTO;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 *
 * @author sky
 * @date 2021年06月16日 11:41 上午
 */
@Component
@Slf4j
@EnableScheduling
public class DelayedMsgPublisher {

    private static final String delayQueueName = "delay_queue_name";

    @Autowired
    private RedissonClient client;

    /**
     * 使用延迟队列发送消息
     * @param  msg 消息体
     * @param ttl 延迟时间
     */
    public void sendMsg(final MessageDTO msg, long ttl) {
        // 延迟队列
        RBlockingQueue<MessageDTO> blockingQueue = client.getBlockingQueue(delayQueueName);

        // 阻塞队列
        RDelayedQueue<MessageDTO> delayedQueue = client.getDelayedQueue(blockingQueue);

        // 往延迟队列发送消息（设置 ttl），相当于延迟了"阻塞队列"中消息的接收
        delayedQueue.offer(msg, ttl, TimeUnit.MILLISECONDS);

        log.info("发送消息，{}", msg);
    }

    @Scheduled(cron = "*/1 * * * * ?")
    public void consumeMsg() throws InterruptedException {
        log.info("准备接收消息");
        RBlockingQueue<MessageDTO> blockingQueue = client.getBlockingQueue(delayQueueName);
        MessageDTO msg = blockingQueue.take();
        log.info("接收到消息了，{}", msg);
    }
}


